package com.etl.dataflow.common.entity;

import lombok.Data;
import lombok.ToString;
import lombok.experimental.Accessors;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 任务统计信息
 *
 * @author dx
 * @since 2022/4/21
 */
@Data
@Accessors(chain = true)
@ToString
public class TaskStatisticInfo {
    /**
     * 源表数据量
     */
    private AtomicLong srcRowNumber = new AtomicLong(-1);

    private AtomicLong redRowNumber = new AtomicLong();
    private AtomicLong processedRowNumber = new AtomicLong();
    private AtomicLong writtenRowNumber = new AtomicLong();

    /**
     * 读缓冲速度: 字节/s
     */
    private AtomicLong readBytesPerSecond = new AtomicLong();
    /**
     * 读缓冲速度: 行/s
     */
    private AtomicLong readRowsPerSecond = new AtomicLong();
    /**
     * 写缓冲速度: 字节/s
     */
    private AtomicLong writeBytesPerSecond = new AtomicLong();
    /**
     * 写缓冲速度: 行/s
     */
    private AtomicLong writeRowsPerSecond = new AtomicLong();
    private AtomicLong readWaitTime = new AtomicLong(0L);
    private AtomicLong readWaitCount = new AtomicLong(0L);
    private AtomicLong writeWaitTime = new AtomicLong(0L);
    private AtomicLong writeWaitCount = new AtomicLong(0L);

    /**
     * 计算每万条写缓冲阻塞时间(ns)
     *
     * @return nanoTime
     */
    public BigDecimal getWriteBufferBlockingTimePer10Kr() {
        if (writeWaitCount.get() == 0) {
            return BigDecimal.ZERO;
        }
        return new BigDecimal(writeWaitTime.get())
                .divide(new BigDecimal(writeWaitCount.get()), RoundingMode.HALF_UP)
                .multiply(new BigDecimal("1e+4"));
    }

    /**
     * 计算每万条读缓冲阻塞时间(ns)
     *
     * @return nanoTime
     */
    public BigDecimal getReadBufferBlockingTimePer10Kr() {
        if (readWaitCount.get() == 0) {
            return BigDecimal.ZERO;
        }
        return new BigDecimal(readWaitTime.get())
                .divide(new BigDecimal(readWaitCount.get()), RoundingMode.HALF_UP)
                .multiply(new BigDecimal("1e+4"));
    }

    public void addAndGetReadWaitTime(long waitTime) {
        this.readWaitTime.addAndGet(waitTime);
        this.readWaitCount.incrementAndGet();
    }

    public void addAndGetWriteWaitTime(long waitTime) {
        this.writeWaitTime.addAndGet(waitTime);
        this.writeWaitCount.incrementAndGet();
    }

    @Data
    @Accessors(chain = true)
    @ToString
    public static class Dto {
        private long srcRowNumber = 0;

        private long redRowNumber = 0;
        private long processedRowNumber = 0;
        private long writtenRowNumber = 0;

        private long readBytesPerSecond = 0;
        private long readRowsPerSecond = 0;

        private long writeBytesPerSecond = 0;
        private long writeRowsPerSecond = 0;

        private long writeBufferBlockingTimePer10Kr = 0;
        private long readBufferBlockingTimePer10Kr = 0;
    }
}
